# Copyright (c) 2019 Presto Labs Pte. Ltd.
# Author: donggu

import logging

import pandas as pd

import coin.strategy.mm.tool.archive_base as abase
from coin.base.datetime_util import to_datetime, to_timestamp_int
from coin.proto.support.coin_order_stateless_pb2 import StatelessOrderEvent
from experimental.prophet.ops.dataframe_feeder import dataframe_feeder

DEFAULT_ORDER_ROOT_DIR = '/remote/iosg/coin/data/og_proto_stateless_log/'


def get_stateless_order_event_df(market_type,
                                 exchange,
                                 start_time,
                                 end_time,
                                 machine,
                                 owner,
                                 event_columns,
                                 state_columns,
                                 order_root_dir=None):

  if order_root_dir is None:
    order_root_dir = DEFAULT_ORDER_ROOT_DIR

  start_ts = to_timestamp_int(start_time)
  end_ts = to_timestamp_int(end_time)

  events = []

  def on_log(timestamp, raw):
    if timestamp < start_ts or timestamp >= end_ts:
      return

    log = StatelessOrderEvent()
    try:
      log.ParseFromString(raw)
      events.append([getattr(log.event, col) for col in event_columns]
                    + [getattr(log.state, col) for col in state_columns])
    except Exception as e:
      print('stateless og log parsing error: ', e)
      return

  abase.run_from_stateless_order_archive(on_log=on_log,
                                         market_type=market_type,
                                         exchange=exchange,
                                         machine=machine,
                                         root_dir=order_root_dir,
                                         owner=owner,
                                         start_time=start_time,
                                         end_time=end_time)

  return pd.DataFrame(events, columns=event_columns + state_columns)


def coin_order_event(market_type,
                     exchange,
                     start_time,
                     end_time,
                     machine,
                     owner,
                     event_columns,
                     state_columns,
                     order_root_dir=None):
  """Get a prophet stream for stateless order events.

  A stateless order event has extra information in addition to OrderEvent such as order_submitted_time and order_cancelled_time.

  Example:
      start_time = datetime.datetime(2019, 8, 15, tzinfo=datetime.timezone.utc)
      end_time = start_time + datetime.timedelta(days=1)
      get_order_event('Futures', 'Bitmex', start_time, end_time, 'strategy-15.eu-west-1',
                                    'xunke33', ['order_price', 'order_qty'], ['order_submitted_time'])

  Arguments:
      market_type {str}
      exchange {str}
      start_time {datetime.datetime}
      end_time {datetime.datetime}
      machine {str}
      owner {str}
      event_columns {list} -- a list of OrderEvent columns. See coin_order_gateway.proto.
      state_columns {list} -- a list of state columns. See coin_order_stateless.proto.

  Returns:
      pandas.DataFrame -- a dataframe with the requested columns and 'timestamp' column.
  """
  if order_root_dir is None:
    order_root_dir = DEFAULT_ORDER_ROOT_DIR

  if 'event_time' not in event_columns:
    event_columns.append('event_time')

  df = get_stateless_order_event_df(market_type,
                                    exchange,
                                    start_time,
                                    end_time,
                                    machine,
                                    owner,
                                    event_columns,
                                    state_columns)

  if len(df) == 0:
    return None

  df.rename(columns={'event_time': 'timestamp'}, inplace=True)

  # strictly increasing timestamp
  for i in range(1, len(df)):
    if df.loc[i, 'timestamp'] <= df.loc[i - 1, 'timestamp']:
      df.loc[i, 'timestamp'] = df.loc[i - 1, 'timestamp'] + 1

  logging.info('Order event from %s to %s',
               to_datetime(int(df['timestamp'].iloc[0])),
               to_datetime(int(df['timestamp'].iloc[-1])))

  stream = dataframe_feeder(df)
  return stream
